package com.lld.im.service.message.server;

import com.alibaba.fastjson.JSONObject;
import com.ecode.imecode.pack.message.ChartMessageAckPack;
import com.ecode.imecode.pack.message.messageReceiveServerAckPack;
import com.ecode.imecode.proto.Message;
import com.lld.im.common.ResponseVO;
import com.lld.im.common.constants.Constants;
import com.lld.im.common.enums.commond.MessageCommand;
import com.lld.im.common.model.UserSession;
import com.lld.im.common.model.message.MessageContent;
import com.lld.im.service.utils.ConversationIdGenerate;
import com.lld.im.service.utils.MessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class P2PMessageServer {

    @Autowired
    private CheckChartServer checkChartServer;

    @Autowired
    private MessageStoreServer messageStoreServer;

    @Autowired
    private StringRedisTemplate redisTemplate;

    private final static ThreadPoolExecutor threadPoolExecutor;

    static {
        //线程池大小 最大大小 存活时间
        threadPoolExecutor = new ThreadPoolExecutor(8, 8, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000)
                , runnable -> {
            Thread thread = new Thread(runnable);
            //设置线程在后台运行
            thread.setDaemon(true);
            return thread;
        });
    }

    @Autowired
    private MessageProducer messageProducer;

    public void process(MessageContent messageContent) {

//        Integer appId = messageContent.getAppId();
//        String fromId = messageContent.getFromId();
//        String toId = messageContent.getToId();
        //前置校验 对于前置校验 可以在tcp层 通过feign进行检验 直接在tcp层拦截请求 节省mq资源
//        ResponseVO responseVO = checkUserStatus(appId, fromId, toId);
//        if (responseVO.isOk()) {
        //TODO 消息幂等性问题 对于重复的消息 服务端暂存消息


        MessageContent cacheMsg = getCacheMsg(messageContent.getAppId(), messageContent.getMessageId());
        //缓存中已经存在 说明是用客户端重发的消息
        if(cacheMsg!=null){
            threadPoolExecutor.execute(()->{
                ack(cacheMsg, ResponseVO.successResponse());
                //  2. IM服务器将数据给发送端的所有在线客户端(数据同步)
                // 3.IM服务器将数据发送目标用户的所有端
                sendMsgToTargetUser(cacheMsg);
                //TODO
                List<UserSession> list = syncSendFromUserAllClient(cacheMsg);
                //TODO 目标用户没有在线客户端 服务器回复ack 如果存在 则交由客户端回复rec-ack
                if (list.isEmpty()) {
                    sendRecAck(cacheMsg);
                }
            });
            return;
        }

        //TODO 对于消息的分发 可以直接交给线程池处理
        //TODO 使用双向ack确认 当A->B B回复一个rec-ack 如果B不在线 由服务端回复serve-ack A再未收到ack之前 都会再指定时间进行重发 最多5分钟

        //TODO 消息有序性 使用redis的increase命令的原子性 对消息顺序进行原子递增
        String key=messageContent.getAppId()+ Constants.SeqConstants.Message+
                ConversationIdGenerate.generateP2PId(messageContent.getFromId(),messageContent.getToId());
        Long seq = redisTemplate.opsForValue().increment(key);
        messageContent.setMessageSequence(seq);
        threadPoolExecutor.execute(() -> {
            // 1.回复ack给发送端
            //消息持久化
            messageStoreServer.storeP2PMessage(messageContent);
            ack(messageContent, ResponseVO.successResponse());
            //  2. IM服务器将数据给发送端的所有在线客户端(数据同步)
            // 3.IM服务器将数据发送目标用户的所有端
            sendMsgToTargetUser(messageContent);
            //TODO
            List<UserSession> list = syncSendFromUserAllClient(messageContent);
            //TODO 目标用户没有在线客户端 服务器回复ack 如果存在 则交由客户端回复rec-ack
            if (list.isEmpty()) {
                sendRecAck(messageContent);
            }
            //
            cacheMsg(messageContent);
        });
//        } else {
        //校验未通过
        // 发送失败 错误ack
//            ack(messageContent, responseVO);
//        }
    }

    private void cacheMsg(MessageContent messageContent) {
        String key=messageContent.getAppId()+Constants.RedissonConfig.MESSAGE_CACHE+ messageContent.getMessageId();
        //默认只保存5min
        redisTemplate.opsForValue().set(key, JSONObject.toJSONString(messageContent),300,TimeUnit.SECONDS);
    }
    
    private MessageContent getCacheMsg(Integer appId,Integer messageId){
        String key=appId+Constants.RedissonConfig.MESSAGE_CACHE+ messageId;
        String cacheMsg = redisTemplate.opsForValue().get(key);
        if(StringUtils.isBlank(cacheMsg)){
            return null;
        }
        return JSONObject.parseObject(cacheMsg,MessageContent.class);
    }


    private void sendRecAck(MessageContent messageContent) {
        //TODO 目标用户在线 交给目标用户确认 不在线 由im服务回复ack
        messageReceiveServerAckPack messageReceiveServerAckPack = new messageReceiveServerAckPack();
        messageReceiveServerAckPack.setServerSend(true);
        messageReceiveServerAckPack.setFromId(messageContent.getFromId());
        messageReceiveServerAckPack.setToId(messageContent.getToId());
        messageReceiveServerAckPack.setMessageSequence(messageReceiveServerAckPack.getMessageSequence());
        messageProducer.sendToTargetSingleUser(messageContent.getFromId(), MessageCommand.MSG_RECIVE_ACK,
                messageContent, messageContent);
    }




    /**
     * 将消息发送给toId的所有客户端
     *
     * @param messageContent
     */
    private void sendMsgToTargetUser(MessageContent messageContent) {
        log.info("message send all toUser client :{}", messageContent.getMessageId());
        messageProducer.sendToAllOnlineUser(messageContent.getToId(), MessageCommand.MSG_P2P,
                messageContent, messageContent.getAppId());
    }


    /**
     * 将消息同步给发送方的所有客户端
     *
     * @param messageContent
     */
    private List<UserSession> syncSendFromUserAllClient(MessageContent messageContent) {
        log.info("message sync all from client :{}", messageContent.getMessageId());
        return messageProducer.sendToUserExceptClient(messageContent.getFromId(), MessageCommand.MSG_P2P
                , messageContent, messageContent);
    }


    /**
     * ack确认是发给发送方的
     *
     * @param messageContent
     */
    private void ack(MessageContent messageContent, ResponseVO responseVO) {
        log.info("message ack :{}", messageContent.getMessageId());
        log.info("message ack status success!");
        //由于tcp层以及完成了前置校验 所有不会有失败的resp
        ChartMessageAckPack chartMessageAckPack = new ChartMessageAckPack();
        chartMessageAckPack.setMessageId(messageContent.getMessageId());
        responseVO.setData(chartMessageAckPack);
        messageProducer.sendToTargetSingleUser(messageContent.getFromId(),
                MessageCommand.MSG_ACK, responseVO, messageContent);
    }

    public ResponseVO checkUserStatus(Integer appId, String fromId, String toId) {
        //判断
        ResponseVO responseVO = checkChartServer.checkUserForbiddenOrProhibitionStatus(appId, fromId, toId);
        if (!responseVO.isOk()) {
            return responseVO;
        }
        responseVO = checkChartServer.checkRelation(appId, fromId, toId);
        return responseVO;
    }
}
